//! # Redis 消息总线适配器
//!
//! 基于 Redis 的消息总线实现，支持发布订阅模式和消息持久化
//!
//! ## 功能特性
//!
//! - **Redis Pub/Sub**：基于 Redis 发布订阅机制
//! - **Redis Streams**：支持消息流和消费者组
//! - **消息持久化**：利用 Redis 持久化特性
//! - **集群支持**：支持 Redis 集群模式
//! - **连接池**：高效的连接管理
//! - **重试机制**：自动重连和错误恢复

use crate::{
    bus::{MessageBus, MessageBusStats, Subscriber, SubscriberStats},
    message::Message,
    subscriber::SubscriberConfig,
    error::{MessageBusError, MessageBusResult},
    router::MessageRouter,
};
use async_trait::async_trait;
use redis::{
    AsyncCommands, Client,
    aio::MultiplexedConnection, streams::StreamReadOptions,
};
use serde::{Deserialize, Serialize};
use std::{
    collections::HashMap,
    sync::{Arc, atomic::{AtomicU64, AtomicUsize, Ordering}},
    time::{Duration, SystemTime, UNIX_EPOCH},
};
use tokio::{
    sync::{RwLock, mpsc},
    task::JoinHandle,
    time::{sleep, timeout},
};
use tracing::{debug, error, info, warn};
use uuid::Uuid;

/// Redis 适配器配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RedisAdapterConfig {
    /// Redis 连接 URL
    pub redis_url: String,
    /// 连接池大小
    pub pool_size: usize,
    /// 连接超时 (秒)
    pub connection_timeout: u64,
    /// 命令超时 (秒)
    pub command_timeout: u64,
    /// 重试次数
    pub max_retries: usize,
    /// 重试间隔 (毫秒)
    pub retry_interval_ms: u64,
    /// 是否使用 Redis Streams
    pub use_streams: bool,
    /// 消费者组名称
    pub consumer_group: Option<String>,
    /// 消费者名称
    pub consumer_name: Option<String>,
    /// 流的最大长度
    pub stream_max_length: Option<u64>,
}

/// Redis 消息总线统计信息
#[derive(Debug, Default)]
struct RedisMessageBusStats {
    total_published: AtomicU64,
    total_received: AtomicU64,
    total_failed: AtomicU64,
    active_subscribers: AtomicUsize,
}

impl Default for RedisAdapterConfig {
    fn default() -> Self {
        Self {
            redis_url: "redis://localhost:6379".to_string(),
            pool_size: 10,
            connection_timeout: 5,
            command_timeout: 5,
            max_retries: 3,
            retry_interval_ms: 1000,
            use_streams: true,
            consumer_group: Some("rustcloud_bus".to_string()),
            consumer_name: Some(format!("consumer_{}", Uuid::new_v4())),
            stream_max_length: Some(10000),
        }
    }
}

/// Redis 消息总线实现
pub struct RedisMessageBus {
    client: Client,
    connection_pool: Arc<RwLock<Vec<MultiplexedConnection>>>,
    config: RedisAdapterConfig,
    stats: Arc<RedisMessageBusStats>,
    router: Arc<MessageRouter>,
    subscribers: Arc<RwLock<HashMap<String, Arc<RedisSubscriber>>>>,
    is_closed: Arc<std::sync::atomic::AtomicBool>,
    tasks: Arc<RwLock<Vec<JoinHandle<()>>>>,
}

impl RedisMessageBus {
    /// 创建新的 Redis 消息总线
    pub async fn new() -> MessageBusResult<Self> {
        let config = RedisAdapterConfig::default();
        Self::with_config(config).await
    }

    /// 使用配置创建 Redis 消息总线
    pub async fn with_config(config: RedisAdapterConfig) -> MessageBusResult<Self> {
        let client = Client::open(config.redis_url.clone())
            .map_err(|e| MessageBusError::connection_error(format!("Redis连接失败: {}", e)))?;

        // 创建连接池
        let mut pool = Vec::new();
        for _ in 0..config.pool_size {
            let conn = client.get_multiplexed_async_connection().await
                .map_err(|e| MessageBusError::connection_error(format!("创建连接失败: {}", e)))?;
            pool.push(conn);
        }

        let bus = Self {
            client,
            connection_pool: Arc::new(RwLock::new(pool)),
            config,
            stats: Arc::new(RedisMessageBusStats::default()),
            router: Arc::new(MessageRouter::new()),
            subscribers: Arc::new(RwLock::new(HashMap::new())),
            is_closed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            tasks: Arc::new(RwLock::new(Vec::new())),
        };

        // 初始化 Redis Streams 消费者组（如果启用）
        if bus.config.use_streams {
            bus.initialize_consumer_groups().await?;
        }

        info!("Redis 消息总线初始化完成");
        Ok(bus)
    }

    /// 初始化消费者组
    async fn initialize_consumer_groups(&self) -> MessageBusResult<()> {
        if let (Some(group), Some(_consumer)) = (&self.config.consumer_group, &self.config.consumer_name) {
            let mut conn = self.get_connection().await?;
            
            // 尝试创建消费者组，如果已存在则忽略错误
            let stream_key = "rustcloud:messages";
            let _: Result<String, redis::RedisError> = conn.xgroup_create_mkstream(
                stream_key, group, "$"
            ).await;
            
            debug!("消费者组 '{}' 初始化完成", group);
        }
        Ok(())
    }

    /// 从连接池获取连接
    async fn get_connection(&self) -> MessageBusResult<MultiplexedConnection> {
        let mut pool = self.connection_pool.write().await;
        
        if let Some(conn) = pool.pop() {
            Ok(conn)
        } else {
            // 如果池为空，创建新连接
            self.client.get_multiplexed_async_connection().await
                .map_err(|e| MessageBusError::connection_error(format!("获取连接失败: {}", e)))
        }
    }

    /// 归还连接到池
    async fn return_connection(&self, conn: MultiplexedConnection) {
        let mut pool = self.connection_pool.write().await;
        pool.push(conn);
    }

    /// 序列化消息
    fn serialize_message(&self, message: &Message) -> MessageBusResult<String> {
        serde_json::to_string(message)
            .map_err(|e| MessageBusError::serialization_error(format!("消息序列化失败: {}", e)))
    }

    /// 反序列化消息
    fn deserialize_message(&self, data: &str) -> MessageBusResult<Message> {
        serde_json::from_str(data)
            .map_err(|e| MessageBusError::deserialization_error(format!("消息反序列化失败: {}", e)))
    }
}

#[async_trait]
impl MessageBus for RedisMessageBus {
    async fn publish(&self, message: Message) -> MessageBusResult<()> {
        if self.is_closed.load(Ordering::Relaxed) {
            return Err(MessageBusError::internal_error("消息总线已关闭"));
        }

        let serialized = self.serialize_message(&message)?;
        let topic = message.topic().to_string();

        let mut conn = self.get_connection().await?;

        if self.config.use_streams {
            // 使用 Redis Streams
            let stream_key = format!("rustcloud:stream:{}", topic);
            let timestamp_str = SystemTime::now().duration_since(UNIX_EPOCH)
                .unwrap().as_millis().to_string();
            let fields = vec![
                ("data", serialized.as_str()),
                ("timestamp", timestamp_str.as_str()),
            ];

            let _: String = conn.xadd(&stream_key, "*", &fields).await
                .map_err(|e| MessageBusError::publish_error(topic.clone(), format!("发布到流失败: {}", e)))?;

            // 设置流的最大长度
            if let Some(max_len) = self.config.stream_max_length {
                let _: u64 = conn.xtrim(&stream_key, redis::streams::StreamMaxlen::Approx(max_len as usize)).await
                    .unwrap_or_default();
            }
        } else {
            // 使用传统 Pub/Sub
            let channel = format!("rustcloud:channel:{}", topic);
            let _: u64 = conn.publish(&channel, &serialized).await
                .map_err(|e| MessageBusError::publish_error(topic.clone(), format!("发布到频道失败: {}", e)))?;
        }

        self.return_connection(conn).await;
        self.stats.total_published.fetch_add(1, Ordering::Relaxed);
        
        debug!("消息发布成功: topic={}", topic);
        Ok(())
    }

    async fn subscribe(&self, pattern: &str) -> MessageBusResult<Box<dyn Subscriber>> {
        if self.is_closed.load(Ordering::Relaxed) {
            return Err(MessageBusError::internal_error("消息总线已关闭"));
        }

        let subscription_id = Uuid::new_v4().to_string();
        let config = SubscriberConfig {
            buffer_size: 1000,
            auto_ack: true,
            max_unacked: 100,
            message_timeout_ms: 5000,
            prefetch_count: 10,
        };

        let subscriber = RedisSubscriber::new(
            self.client.clone(),
            self.config.clone(),
            subscription_id.clone(),
            pattern.to_string(),
            config,
            self.stats.clone(),
        ).await?;

        let subscriber_arc = Arc::new(subscriber);
        
        {
            let mut subs = self.subscribers.write().await;
            subs.insert(subscription_id.clone(), subscriber_arc.clone());
        }

        self.stats.active_subscribers.fetch_add(1, Ordering::Relaxed);
        
        info!("创建订阅者: pattern={}, id={}", pattern, subscription_id);
        Ok(Box::new(subscriber_arc.as_ref().clone()))
    }

    async fn unsubscribe(&self, subscription_id: &str) -> MessageBusResult<()> {
        let mut subs = self.subscribers.write().await;
        if let Some(subscriber) = subs.remove(subscription_id) {
            subscriber.close().await?;
            self.stats.active_subscribers.fetch_sub(1, Ordering::Relaxed);
            info!("取消订阅: id={}", subscription_id);
        }
        Ok(())
    }

    async fn list_topics(&self) -> MessageBusResult<Vec<String>> {
        let mut conn = self.get_connection().await?;
        
        let keys: Vec<String> = if self.config.use_streams {
            conn.keys("rustcloud:stream:*").await
                .map_err(|e| MessageBusError::internal_error(format!("获取流列表失败: {}", e)))?
        } else {
            conn.keys("rustcloud:channel:*").await
                .map_err(|e| MessageBusError::internal_error(format!("获取频道列表失败: {}", e)))?
        };

        let prefix = if self.config.use_streams {
            "rustcloud:stream:"
        } else {
            "rustcloud:channel:"
        };

        let topics = keys.into_iter()
            .filter_map(|key| key.strip_prefix(prefix).map(|s| s.to_string()))
            .collect();

        self.return_connection(conn).await;
        Ok(topics)
    }

    async fn stats(&self) -> MessageBusResult<MessageBusStats> {
        Ok(MessageBusStats {
            total_published: self.stats.total_published.load(Ordering::Relaxed),
            total_received: self.stats.total_received.load(Ordering::Relaxed),
            total_failed: self.stats.total_failed.load(Ordering::Relaxed),
            active_subscribers: self.stats.active_subscribers.load(Ordering::Relaxed),
            active_topics: 0, // TODO: 实现主题计数
            queue_length: 0, // TODO: 实现队列长度计数
            uptime_seconds: SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_secs(),
            messages_per_second: 0.0, // TODO: 实现吗秒速率计算
            average_message_size: 0, // TODO: 实现平均消息大小计算
            memory_usage: 0, // TODO: 实现内存使用量计算
        })
    }

    async fn health_check(&self) -> MessageBusResult<bool> {
        if self.is_closed.load(Ordering::Relaxed) {
            return Ok(false);
        }

        let mut conn = self.get_connection().await?;
        let result: String = redis::cmd("PING").query_async(&mut conn).await
            .map_err(|e| MessageBusError::connection_error(format!("健康检查失败: {}", e)))?;
        
        self.return_connection(conn).await;
        Ok(result == "PONG")
    }

    async fn close(&self) -> MessageBusResult<()> {
        if self.is_closed.swap(true, Ordering::Relaxed) {
            return Ok(()); // 已经关闭
        }

        info!("正在关闭 Redis 消息总线...");

        // 关闭所有订阅者
        let mut subs = self.subscribers.write().await;
        for (_, subscriber) in subs.drain() {
            let _ = subscriber.close().await;
        }

        // 停止所有后台任务
        let mut tasks = self.tasks.write().await;
        for task in tasks.drain(..) {
            task.abort();
        }

        info!("Redis 消息总线已关闭");
        Ok(())
    }

    async fn subscriber_count(&self, topic: &str) -> MessageBusResult<usize> {
        let subs = self.subscribers.read().await;
        let mut count = 0;
        for subscriber in subs.values() {
            if subscriber.is_active() && subscriber.pattern() == topic {
                count += 1;
            }
        }
        Ok(count)
    }

    async fn has_subscribers(&self, topic: &str) -> MessageBusResult<bool> {
        let count = self.subscriber_count(topic).await?;
        Ok(count > 0)
    }
}

/// Redis 订阅者实现
#[derive(Clone)]
pub struct RedisSubscriber {
    subscription_id: String,
    pattern: String,
    config: SubscriberConfig,
    redis_config: RedisAdapterConfig,
    client: Client,
    stats: Arc<RedisMessageBusStats>,
    receiver: Arc<RwLock<Option<mpsc::UnboundedReceiver<Message>>>>,
    is_closed: Arc<std::sync::atomic::AtomicBool>,
    task_handle: Arc<RwLock<Option<JoinHandle<()>>>>,
}

impl RedisSubscriber {
    async fn new(
        client: Client,
        redis_config: RedisAdapterConfig,
        subscription_id: String,
        pattern: String,
        config: SubscriberConfig,
        bus_stats: Arc<RedisMessageBusStats>,
    ) -> MessageBusResult<Self> {
        let (tx, rx) = mpsc::unbounded_channel();
        
        let subscriber = Self {
            subscription_id,
            pattern,
            config,
            redis_config: redis_config.clone(),
            client: client.clone(),
            stats: bus_stats,
            receiver: Arc::new(RwLock::new(Some(rx))),
            is_closed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
            task_handle: Arc::new(RwLock::new(None)),
        };

        // 启动消息接收任务
        let task = subscriber.start_receiving_task(tx).await?;
        *subscriber.task_handle.write().await = Some(task);

        Ok(subscriber)
    }

    async fn start_receiving_task(&self, tx: mpsc::UnboundedSender<Message>) -> MessageBusResult<JoinHandle<()>> {
        let client = self.client.clone();
        let pattern = self.pattern.clone();
        let redis_config = self.redis_config.clone();
        let stats = self.stats.clone();
        let is_closed = self.is_closed.clone();

        let task = tokio::spawn(async move {
            let mut conn = match client.get_multiplexed_async_connection().await {
                Ok(conn) => conn,
                Err(e) => {
                    error!("订阅者连接失败: {}", e);
                    return;
                }
            };

            if redis_config.use_streams {
                // 使用 Redis Streams
                Self::receive_from_streams(&mut conn, &pattern, &redis_config, &tx, &stats, &is_closed).await;
            } else {
                // 使用传统 Pub/Sub
                Self::receive_from_pubsub(&mut conn, &pattern, &tx, &stats, &is_closed).await;
            }
        });

        Ok(task)
    }

    async fn receive_from_streams(
        conn: &mut MultiplexedConnection,
        pattern: &str,
        redis_config: &RedisAdapterConfig,
        tx: &mpsc::UnboundedSender<Message>,
        stats: &Arc<RedisMessageBusStats>,
        is_closed: &Arc<std::sync::atomic::AtomicBool>,
    ) {
        let stream_key = format!("rustcloud:stream:{}", pattern);
        let group = redis_config.consumer_group.as_ref().unwrap();
        let consumer = redis_config.consumer_name.as_ref().unwrap();

        while !is_closed.load(Ordering::Relaxed) {
            let opts = StreamReadOptions::default()
                .group(group, consumer)
                .count(10)
                .block(1000);

            match conn.xread_options::<&str, &str, redis::streams::StreamReadReply>(&[&stream_key], &[">"], &opts).await {
                Ok(reply) => {
                    for stream_id in reply.keys {
                        for stream_msg in stream_id.ids {
                            if let Some(data) = stream_msg.map.get("data") {
                                if let redis::Value::Data(bytes) = data {
                                    if let Ok(data_str) = std::str::from_utf8(bytes) {
                                        if let Ok(message) = serde_json::from_str::<Message>(data_str) {
                                            if tx.send(message).is_err() {
                                                warn!("发送消息到通道失败，接收者可能已关闭");
                                                return;
                                            }
                                            stats.total_received.fetch_add(1, Ordering::Relaxed);

                                            // 确认消息
                                            let _: Result<u64, redis::RedisError> = conn.xack(&stream_key, group, &[&stream_msg.id]).await;
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
                Err(e) => {
                    if !is_closed.load(Ordering::Relaxed) {
                        error!("从 Redis Stream 读取失败: {}", e);
                        stats.total_failed.fetch_add(1, Ordering::Relaxed);
                        sleep(Duration::from_millis(redis_config.retry_interval_ms)).await;
                    }
                }
            }
        }
    }

    async fn receive_from_pubsub(
        _conn: &mut MultiplexedConnection,
        _pattern: &str,
        _tx: &mpsc::UnboundedSender<Message>,
        _stats: &Arc<RedisMessageBusStats>,
        is_closed: &Arc<std::sync::atomic::AtomicBool>,
    ) {
        // 这里需要使用 PubSub 连接，但为了简化暂时使用轮询方式
        // 在实际实现中应该使用 redis::aio::PubSub
        while !is_closed.load(Ordering::Relaxed) {
            sleep(Duration::from_millis(100)).await;
            // TODO: 实现真正的 Pub/Sub 订阅
        }
    }
}

#[async_trait]
impl Subscriber for RedisSubscriber {
    fn subscription_id(&self) -> &str {
        &self.subscription_id
    }

    fn pattern(&self) -> &str {
        &self.pattern
    }

    fn is_active(&self) -> bool {
        !self.is_closed.load(Ordering::Relaxed)
    }

    async fn receive(&self) -> MessageBusResult<Option<Message>> {
        self.receive_timeout(5000).await
    }

    async fn try_receive(&self) -> MessageBusResult<Option<Message>> {
        if self.is_closed.load(Ordering::Relaxed) {
            return Err(MessageBusError::internal_error("订阅已关闭"));
        }

        let mut receiver_guard = self.receiver.write().await;
        if let Some(ref mut receiver) = receiver_guard.as_mut() {
            match receiver.try_recv() {
                Ok(message) => Ok(Some(message)),
                Err(mpsc::error::TryRecvError::Empty) => Ok(None),
                Err(mpsc::error::TryRecvError::Disconnected) => {
                    Err(MessageBusError::internal_error("订阅已关闭"))
                }
            }
        } else {
            Err(MessageBusError::internal_error("订阅已关闭"))
        }
    }

    async fn receive_timeout(&self, timeout_ms: u64) -> MessageBusResult<Option<Message>> {
        if self.is_closed.load(Ordering::Relaxed) {
            return Err(MessageBusError::internal_error("订阅已关闭"));
        }

        let mut receiver_guard = self.receiver.write().await;
        if let Some(ref mut receiver) = receiver_guard.as_mut() {
            match timeout(Duration::from_millis(timeout_ms), receiver.recv()).await {
                Ok(Some(message)) => Ok(Some(message)),
                Ok(None) => Err(MessageBusError::internal_error("订阅已关闭")),
                Err(_) => Ok(None), // 超时
            }
        } else {
            Err(MessageBusError::internal_error("订阅已关闭"))
        }
    }

    async fn ack(&self, _message: &Message) -> MessageBusResult<()> {
        // Redis Streams 的 ACK 在接收时自动处理
        Ok(())
    }

    async fn nack(&self, _message: &Message, _requeue: bool) -> MessageBusResult<()> {
        // Redis Streams 的 NACK 处理
        Ok(())
    }

    async fn stats(&self) -> MessageBusResult<SubscriberStats> {
        Ok(SubscriberStats {
            total_received: 0,
            total_acked: 0,
            total_nacked: 0,
            queue_length: 0,
            average_processing_time_ms: 0.0,
            last_received_at: None,
            subscribed_at: chrono::Utc::now(),
        })
    }

    async fn close(&self) -> MessageBusResult<()> {
        if self.is_closed.swap(true, Ordering::Relaxed) {
            return Ok(()); // 已经关闭
        }

        // 停止接收任务
        if let Some(task) = self.task_handle.write().await.take() {
            task.abort();
        }

        // 关闭接收器
        *self.receiver.write().await = None;

        debug!("Redis 订阅者已关闭: {}", self.subscription_id);
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    async fn create_test_bus() -> RedisMessageBus {
        let config = RedisAdapterConfig {
            redis_url: "redis://localhost:6379".to_string(),
            pool_size: 2,
            use_streams: true,
            ..Default::default()
        };
        
        RedisMessageBus::with_config(config).await.unwrap()
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_bus_creation() {
        let bus = create_test_bus().await;
        assert!(bus.health_check().await.unwrap());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_publish_subscribe() {
        let bus = create_test_bus().await;
        
        let subscriber = bus.subscribe("test.topic").await.unwrap();
        
        let message = Message::new("test.topic", json!({"data": "test"}));
        bus.publish(message).await.unwrap();
        
        let received = subscriber.receive_timeout(1000).await.unwrap();
        assert!(received.is_some());
        
        let received_msg = received.unwrap();
        assert_eq!(received_msg.topic(), "test.topic");
        assert_eq!(received_msg.payload()["data"], "test");
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_multiple_subscribers() {
        let bus = create_test_bus().await;
        
        let sub1 = bus.subscribe("multi.test").await.unwrap();
        let sub2 = bus.subscribe("multi.test").await.unwrap();
        
        let message = Message::new("multi.test", json!({"id": 1}));
        bus.publish(message).await.unwrap();
        
        let msg1 = sub1.receive_timeout(1000).await.unwrap();
        let msg2 = sub2.receive_timeout(1000).await.unwrap();
        
        assert!(msg1.is_some());
        assert!(msg2.is_some());
    }

    #[tokio::test]
    #[ignore] // 需要运行 Redis 服务器
    async fn test_redis_stats() {
        let bus = create_test_bus().await;
        
        let message = Message::new("stats.test", json!({"test": true}));
        bus.publish(message).await.unwrap();
        
        let stats = bus.stats().await.unwrap();
        assert!(stats.total_published > 0);
    }
}